package com.navinfo.data.esper.cep.linkid;


import com.espertech.esper.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;

public class LinkidMain {
    private static final Logger log = LoggerFactory.getLogger(LinkidMain.class);

    public static void main(String[] args) throws IOException, InterruptedException {
        // Configure engine with event names to make the statements more readable.
        // This could also be done in a configuration file.
        Configuration configuration = new Configuration();
        configuration.addEventType("TerminalLink", TerminalLink.class.getName());

        // Get engine instance
        EPServiceProvider epService = EPServiceProviderManager.getProvider("linkid", configuration);

        log.info("making test data begin......");
        //创建测试数据集
        Map<Integer, List<Integer>> map = new HashMap<Integer, List<Integer>>();
        for (int i = 700001; i < 901100; i++) {
            List<Integer> li = new ArrayList<Integer>();
            for (int j = 1; j < 11; j++) {
                li.add(j+901099-i);
            }
            map.put(i,li);
        }

        List<Integer> li = new ArrayList<Integer>();
        for (int j = 1; j < 11; j++) {
            li.add(j);
        }
        map.put(901100,li);

        List<Integer> roadLinkids=new ArrayList<Integer>();
        for (int j = 1; j < 6; j++) {
            roadLinkids.add(j);
        }

        log.info("making test data end......");

        List<Integer> tidList=new ArrayList<Integer>();
        //Clear engine statement
//        epService.getEPAdministrator().destroyAllStatements();
//        epService.removeAllStatementStateListeners();
        //Set up statements
        //a=AppleEvent -> b=BananaEvent(b.price = a.price) // equals: a=AppleEvent -> b=BananaEvent(price = a.price)
        String epl1 = "create context TerminalLinkByTid partition by tid from TerminalLink";
//        String epl2 = "context TerminalLinkByTid select context.id,context.name,context.key1 from pattern[(([:1] (a0=TerminalLink(linkid=1)) until (b0=TerminalLink(linkid=1))) -> ([:1] (a=TerminalLink(linkid=1)) until (b=TerminalLink(linkid=2))) -> ([:1] (c=TerminalLink(linkid=1)) until (d=TerminalLink(linkid=3))))]";


        epService.getEPAdministrator().createEPL(epl1);
//        EPStatement state = admin.createEPL(epl2);
//        state.addListener(new ContextPropertiesListener4());
//        StringBuffer pattern = new StringBuffer().append("(([:1] (a0=TerminalLink(linkid=1)) until (b0=TerminalLink(linkid=1))) -> ([:1] (a=TerminalLink(linkid=1 and a.tid=a0.tid)) until (b=TerminalLink(linkid=2 and b.tid=a0.tid))) -> ([:1] (c=TerminalLink(linkid=1 and c.tid=a0.tid)) until (d=TerminalLink(linkid=3 and d.tid=a0.tid))))");
//            StringBuffer pattern = new StringBuffer().append("(");
//        StringBuffer pattern = new StringBuffer().append("a=TerminalLink(linkid=1) -> b=TerminalLink(linkid=2 and b.tid = a.tid)");

        StringBuffer pattern = new StringBuffer().append("context TerminalLinkByTid select context.id,context.name,context.key1 from pattern[(");
        int roadlinkid = 0;
        if (roadLinkids.size()>0) {
                for(int z=0;z<roadLinkids.size();z++){
                    roadlinkid = roadLinkids.get(z);
                    if (z==(roadLinkids.size()-1)) {
                        pattern.append("([:1] (TerminalLink(linkid=").append(roadlinkid).append(")) until (TerminalLink(linkid=").append(roadlinkid).append("))))]");
                    } else {
                        pattern.append("([:1] (TerminalLink(linkid=").append(roadlinkid).append(")) until (TerminalLink(linkid=").append(roadlinkid).append("))) -> ");
                    }

                }
        }
        EPStatement statement = epService.getEPAdministrator().createEPL(pattern.toString());
        LinkidPatternListener2 linkidPatternListener2 = new LinkidPatternListener2();
        statement.addListener(linkidPatternListener2);

        for (int i = 700001; i <= 901100; i++) {
            if (i%10000==0) {
                log.info("patterning......:"+i);
            }



            if (roadLinkids.size()>0) {
//                for(int z=0;z<roadLinkids.size();z++){
//                    roadlinkid = roadLinkids.get(z);
//                    if (z==(roadLinkids.size()-1)) {
//                        pattern.append("([:1] (TerminalLink(linkid=").append(roadlinkid).append(")) until (b=TerminalLink(linkid=").append(roadlinkid).append(")))) and TerminalLink(tid=").append(i).append(")");
//                    } else {
//                        pattern.append("([:1] (TerminalLink(linkid=").append(roadlinkid).append(")) until (TerminalLink(linkid=").append(roadlinkid).append("))) -> ");
//                    }
//
//                }


                EPRuntime runtime = epService.getEPRuntime();

                for (int j = 0; j < map.get(i).size(); j++) {
                    TerminalLink tl = new TerminalLink();
                    tl.setTid(i);
                    tl.setLinkid(map.get(i).get(j));
                    runtime.sendEvent(tl);
                }
                if (linkidPatternListener2.getLastEvent()!=null) {
                    EventBean event = linkidPatternListener2.getLastEvent();
//                    log.info("context.tid " + event.get("key1") + ", context.id " + event.get("id") + ", context.name " + event.get("name"));

                    tidList.add((int)event.get("key1"));
//                    log.info("--------------------b query returned: " + event.get("key1"));
                }

            }

        }

        Iterator<Integer> it2 = tidList.iterator();
        while (it2.hasNext()) {
            log.info(it2.next() + " ");
        }

        epService.destroy();
    }
}
